class RepoSpec(BaseModel):
    repo_url: str                          # HTTPS or SSH git URL
    repo_path: str = ""                    # Absolute path; auto-derived if empty
    role: Literal["primary", "dependency"] = "primary"
    branch: str = ""                       # Checkout this branch; empty = repo default
    sparse_paths: list[str] = []           # If non-empty, enable sparse-checkout
    mount_point: str = ""                  # Subdirectory name in workspace root; auto-derived from repo name if empty
    create_pr: bool = True                 # Whether to create a GitHub PR for this repo

class WorkspaceRepo(BaseModel):
    repo_name: str             # Derived from URL (last path segment, no .git)
    repo_url: str
    role: Literal["primary", "dependency"]
    absolute_path: str         # Where repo is on disk
    branch: str                # Checked-out branch after clone
    sparse_paths: list[str]
    create_pr: bool
    git_init_result: dict | None = None   # Populated after git_init per repo

class WorkspaceManifest(BaseModel):
    workspace_root: str           # Absolute path to per-build workspace dir
    repos: list[WorkspaceRepo]    # One entry per cloned repo
    primary_repo_name: str        # repo_name of the designated primary

class RepoPRResult(BaseModel):
    repo_name: str
    repo_url: str
    success: bool
    pr_url: str = ""
    pr_number: int = 0
    error_message: str = ""

repos: list[RepoSpec] = []   # Empty = use legacy repo_url / repo_path

@property
def primary_repo(self) -> RepoSpec | None:
    """Returns the RepoSpec with role='primary', or None if repos is empty."""

# NEW
pr_results: list[RepoPRResult] = []

# DEPRECATED alias (kept for backward compat)
@property
def pr_url(self) -> str:
    """Returns first successful PR URL or empty string. Deprecated; use pr_results."""
    for r in self.pr_results:
        if r.success:
            return r.pr_url
    return ""

# NEW
workspace_manifest: dict | None = None   # Serialized WorkspaceManifest

# NEW optional field
target_repo: str = ""   # repo_name this issue's changes land in; empty = primary

# NEW optional field
repo_name: str = ""     # Which repo was modified; empty = primary

# NEW optional field
repo_name: str = ""     # Populated from run_git_init's repo_name parameter

# NEW optional field
repo_name: str = ""     # Which repo this merge result covers

async def _clone_repos(
    cfg: BuildConfig,
    artifacts_dir: str,
) -> WorkspaceManifest:
    """
    Clone all repos in cfg.repos concurrently.
    Raises RuntimeError if any clone fails.
    Returns WorkspaceManifest with absolute_path populated for each repo.
    """

def workspace_context_block(manifest: WorkspaceManifest) -> str:
    """
    Returns empty string if manifest.repos has exactly one entry.
    Returns markdown workspace layout section for multi-repo manifests.
    """

pr_results: list[RepoPRResult] = []
for repo in manifest.repos:
    if not repo.create_pr:
        continue
    result = await run_github_pr(
        repo_path=repo.absolute_path,
        integration_branch=repo.git_init_result["integration_branch"],
        base_branch=...,
        goal=goal,
        completed_issues=issues_for_repo(repo.repo_name),
        accumulated_debt=...,
        artifacts_dir=artifacts_dir,
        all_pr_results=pr_results,
    )
    pr_results.append(RepoPRResult(...))

async def _clone_repos(cfg: BuildConfig, artifacts_dir: str) -> WorkspaceManifest

def workspace_context_block(manifest: WorkspaceManifest) -> str

def workspace_setup_task_prompt(
    ...,
    workspace_manifest: WorkspaceManifest | None = None,  # NEW keyword param
) -> str

def github_pr_task_prompt(
    ...,
    all_pr_results: list[RepoPRResult] | None = None,  # NEW keyword param
) -> str

import re

def _derive_repo_name(url: str) -> str:
    """Extract repo name from a git URL.

    Examples:
        'https://github.com/org/my-project.git' -> 'my-project'
        'git@github.com:org/my-project.git'     -> 'my-project'
        'https://github.com/org/repo'           -> 'repo'
    """
    match = re.search(r"/([^/]+?)(?:\.git)?$", url.rstrip("/"))
    return match.group(1) if match else "repo"

from swe_af.execution.schemas import _derive_repo_name as _repo_name_from_url

from pydantic import BaseModel, ConfigDict, field_validator

class RepoSpec(BaseModel):
    """Input specification for one repository in a multi-repo build."""

    model_config = ConfigDict(extra="forbid")

    repo_url: str = ""
    """GitHub/git HTTPS or SSH URL. Required when repo_path is empty."""

    repo_path: str = ""
    """Absolute path to an already-cloned repo. Skips clone when set."""

    role: str
    """'primary' or 'dependency'. Exactly one repo must be 'primary'."""

    branch: str = ""
    """Branch to checkout. Empty string means use the remote default branch."""

    sparse_paths: list[str] = []
    """For sparse checkout. Empty list means full checkout."""

    mount_point: str = ""
    """Workspace subdirectory name override. Defaults to name derived from
    repo_url. Must be unique across all repos in a BuildConfig."""

    create_pr: bool = True
    """Whether to create a PR for this repo after the build."""

    @field_validator("repo_url")
    @classmethod
    def _validate_url(cls, v: str) -> str:
        if not v:
            return v  # repo_path may be used instead
        valid_prefixes = ("https://", "http://", "git@", "ssh://")
        if not any(v.startswith(p) for p in valid_prefixes):
            raise ValueError(
                f"repo_url must be an https://, http://, git@, or ssh:// URL; got {v!r}"
            )
        return v

    @field_validator("role")
    @classmethod
    def _validate_role(cls, v: str) -> str:
        if v not in ("primary", "dependency"):
            raise ValueError(f"role must be 'primary' or 'dependency'; got {v!r}")
        return v

class WorkspaceRepo(BaseModel):
    """State of a single repository that has been cloned into the workspace."""

    model_config = ConfigDict(frozen=False)
    """Mutable so _init_all_repos can backfill git_init_result in-place.
    Resolves B-NEW-4: without frozen=False, Pydantic v2 raises TypeError on
    ws_repo.git_init_result = result assignment."""

    repo_name: str
    """Canonical name, derived from repo_url or mount_point. E.g. 'myrepo'."""

    repo_url: str
    """Original git URL (empty if repo_path-only spec)."""

    role: str
    """'primary' or 'dependency'."""

    absolute_path: str
    """Absolute filesystem path to the cloned repo root."""

    branch: str
    """Actual checked-out branch name (resolved post-clone via git rev-parse)."""

    sparse_paths: list[str] = []
    """Sparse checkout paths, or empty for full checkout."""

    create_pr: bool = True
    """Whether a PR should be created for this repo."""

    git_init_result: dict | None = None
    """Populated by _init_all_repos after run_git_init completes.
    Contains GitInitResult.model_dump() for this repo."""

class WorkspaceManifest(BaseModel):
    """Snapshot of all repositories cloned for a multi-repo build."""

    workspace_root: str
    """Absolute path to the parent directory containing all repo subdirectories."""

    repos: list[WorkspaceRepo]
    """All repos in this build. Always at least one entry."""

    primary_repo_name: str
    """repo_name of the primary repo. Guaranteed to match one entry in repos."""

    @property
    def primary_repo(self) -> "WorkspaceRepo":
        """Return the primary WorkspaceRepo. Always present."""
        for r in self.repos:
            if r.repo_name == self.primary_repo_name:
                return r
        raise RuntimeError(
            f"primary_repo_name={self.primary_repo_name!r} not found in manifest"
        )

class RepoPRResult(BaseModel):
    """Result of creating a PR for one repository."""

    repo_name: str
    repo_url: str
    success: bool
    pr_url: str = ""
    pr_number: int = 0
    error_message: str = ""

# New field — add after existing field declarations:
repos: list["RepoSpec"] = []
"""Multi-repo list. Empty means single-repo mode (uses repo_url legacy field).
Populated by the model_validator from legacy repo_url when repos is empty."""

# New model_validator — add after existing _validate_v2_keys:
@model_validator(mode="after")
def _normalize_repos(self) -> "BuildConfig":
    """Normalize legacy repo_url/repo_path into self.repos and validate the list.

    Validation order (stops at first failure):
    1. Mutual exclusion: repo_url + non-empty repos simultaneously -> ValueError
    2. Normalization: empty repos + non-empty repo_url -> synthesize single RepoSpec
    3. Empty passthrough: repos still empty -> return (deferred to build())
    4. Exactly one primary -> ValueError if 0 or 2+
    5. No duplicate repo_url values -> ValueError
    6. No duplicate resolved names -> ValueError
    7. Backfill self.repo_url from primary if currently empty
    """
    caller_set_repos = bool(self.__pydantic_fields_set__ & {"repos"})
    caller_set_url = bool(self.__pydantic_fields_set__ & {"repo_url"})
    if caller_set_url and caller_set_repos and self.repos:
        raise ValueError(
            "Cannot set both repo_url and repos simultaneously. "
            "Use repos=[RepoSpec(...)] for multi-repo builds."
        )

    if not self.repos and self.repo_url:
        self.repos = [RepoSpec(
            repo_url=self.repo_url,
            role="primary",
            create_pr=self.enable_github_pr,
        )]

    if not self.repos:
        return self  # deferred — build() validates repo_path is provided

    primaries = [r for r in self.repos if r.role == "primary"]
    if len(primaries) != 1:
        raise ValueError(
            f"Exactly one repo must have role='primary'; found {len(primaries)}"
        )

    urls = [r.repo_url for r in self.repos if r.repo_url]
    if len(urls) != len(set(urls)):
        raise ValueError("Duplicate repo_url values are not allowed in repos list.")

    def _get_name(spec: "RepoSpec") -> str:
        if spec.mount_point:
            return spec.mount_point
        if spec.repo_url:
            return _derive_repo_name(spec.repo_url)
        return os.path.basename(spec.repo_path.rstrip("/")) if spec.repo_path else "repo"

    names = [_get_name(r) for r in self.repos]
    if len(names) != len(set(names)):
        raise ValueError(
            f"Duplicate repo names/mount_points in repos: {names}. "
            "Set mount_point explicitly to disambiguate."
        )

    if not self.repo_url and primaries:
        # Use object.__setattr__ to bypass Pydantic's immutability on assignment
        object.__setattr__(self, "repo_url", primaries[0].repo_url)

    return self

@property
def primary_repo(self) -> "RepoSpec | None":
    """Return the primary RepoSpec, or None if repos is empty."""
    for r in self.repos:
        if r.role == "primary":
            return r
    return None

# REMOVE: pr_url: str = ""
# ADD:
pr_results: list["RepoPRResult"] = []
"""Per-repo PR creation results. Empty list for builds without PR creation."""

@property
def pr_url(self) -> str:
    """Backward-compat alias: returns first successful PR URL, or empty string."""
    for r in self.pr_results:
        if r.success and r.pr_url:
            return r.pr_url
    return ""

def model_dump(self, **kwargs) -> dict:
    """Override to inject computed pr_url property into serialized dict."""
    d = super().model_dump(**kwargs)
    d["pr_url"] = self.pr_url
    return d

# ADD to DAGState:
workspace_manifest: dict | None = None
"""WorkspaceManifest.model_dump() for multi-repo builds.
None for single-repo builds (backward compat).
Stored as dict (not WorkspaceManifest) so checkpoint JSON serialization works."""

# ADD to CoderResult:
repo_name: str = ""
"""Name of the repo this coder ran in. Empty for single-repo builds."""

# ADD to GitInitResult:
repo_name: str = ""
"""Name of the repo this git_init ran for. Empty for single-repo builds."""

# ADD to MergeResult:
repo_name: str = ""
"""Name of the repo this merge ran for. Empty for single-repo builds."""

# ADD to IssueResult:
repo_name: str = ""
"""Name of the repo this issue was coded in.
Propagated from CoderResult.repo_name by coding_loop.py when recording
coding results. Backfilled from issue['target_repo'] by dag_executor.py
in _execute_level if CoderResult.repo_name is empty.
Used by _merge_level_branches to group branches by repo."""

# ADD to PlannedIssue:
target_repo: str = ""
"""Repo name this issue should be coded in.
Empty means primary repo (backward compat).
Set by sprint planner in multi-repo builds via sprint_planner_task_prompt."""

"""Shared prompt utilities for multi-repo context injection."""
from __future__ import annotations

from swe_af.execution.schemas import WorkspaceManifest


def workspace_context_block(manifest: "WorkspaceManifest | None") -> str:
    """Return a markdown workspace context block for injection into prompts.

    Returns empty string ("") when:
    - manifest is None
    - manifest.repos has exactly one entry (single-repo build)

    For multi-repo builds (len(repos) > 1), returns:

        ## Workspace Repositories
        | Repo | Role | Path | Branch |
        |------|------|------|--------|
        | api (PRIMARY) | primary | /tmp/api | main |
        | lib | dependency | /tmp/lib | develop |

    AC-14: single-repo manifest -> returns ""
    AC-15: multi-repo manifest -> returns string containing repo names and paths

    Note: The PRD §7.1 format spec shows a Branch column; we include it here.
    AC-15 does not test for branch presence but the column is present per spec.
    """
    if manifest is None or len(manifest.repos) <= 1:
        return ""

    lines: list[str] = [
        "## Workspace Repositories",
        "| Repo | Role | Path | Branch |",
        "|------|------|------|--------|",
    ]
    for repo in manifest.repos:
        if repo.repo_name == manifest.primary_repo_name:
            label = f"{repo.repo_name} (PRIMARY)"
        else:
            label = repo.repo_name
        lines.append(
            f"| {label} | {repo.role} | {repo.absolute_path} | {repo.branch} |"
        )

    return "\n".join(lines)

from swe_af.execution.schemas import WorkspaceManifest
from swe_af.prompts._utils import workspace_context_block

def pm_task_prompt(
    *,
    goal: str,
    repo_path: str,
    prd_path: str,
    additional_context: str = "",
    workspace_manifest: "WorkspaceManifest | None" = None,
) -> str:
    """Task prompt for the product manager agent.

    Returns the same content as product_manager_prompts()[1] with optional
    workspace context block injected after the Repository section.
    """
    context_block = ""
    if additional_context:
        context_block = f"\n## Additional Context\n{additional_context}\n"

    ws_block = workspace_context_block(workspace_manifest)
    ws_section = f"\n{ws_block}\n" if ws_block else ""

    return (
        f"## Goal\n{goal}\n\n"
        f"## Repository\n{repo_path}\n"
        f"{context_block}"
        f"{ws_section}"
        "## How Your PRD Will Be Used\n\n"
        "1. An architect designs the technical solution from your PRD\n"
        "2. A sprint planner decomposes into independent issues with a dependency graph\n"
        "3. Issues at the same dependency level execute IN PARALLEL by isolated agents\n"
        "4. A QA agent verifies each acceptance criterion LITERALLY by running commands\n\n"
        "Write acceptance criteria as test assertions, not human briefings.\n\n"
        "## Your Mission\n\n"
        "Produce a PRD for this goal. Read the codebase first — understand the current\n"
        "state deeply before defining what needs to change.\n\n"
        f"Write your full PRD to: {prd_path}\n\n"
        "The bar: an engineering team of autonomous agents can execute this PRD without\n"
        "asking a single clarifying question.\n"
    )

from swe_af.execution.schemas import WorkspaceManifest
from swe_af.prompts._utils import workspace_context_block

def architect_task_prompt(
    *,
    prd: "PRD",
    repo_path: str,
    prd_path: str,
    architecture_path: str,
    feedback: "str | None" = None,
    workspace_manifest: "WorkspaceManifest | None" = None,
) -> str:
    """Task prompt for the architect agent.

    Builds same content as architect_prompts()[1] with optional workspace
    context injected after the Repository section.
    """
    ac_formatted = "\n".join(f"- {c}" for c in prd.acceptance_criteria)
    must_have = "\n".join(f"- {m}" for m in prd.must_have)
    out_of_scope = "\n".join(f"- {o}" for o in prd.out_of_scope)

    feedback_block = ""
    if feedback:
        feedback_block = (
            "\n## Revision Feedback from Tech Lead\n"
            "The previous architecture was reviewed and needs revision:\n"
            f"{feedback}\n"
            "Address these concerns directly.\n"
        )

    ws_block = workspace_context_block(workspace_manifest)
    ws_section = f"\n{ws_block}\n" if ws_block else ""

    return (
        f"## Product Requirements\n{prd.validated_description}\n\n"
        f"## Acceptance Criteria\n{ac_formatted}\n\n"
        f"## Scope\n- Must have:\n{must_have}\n- Out of scope:\n{out_of_scope}\n\n"
        f"## Repository\n{repo_path}\n"
        f"{ws_section}"
        f"The full PRD is at: {prd_path}\n"
        f"{feedback_block}"
        "## Your Mission\n\n"
        "Design the technical architecture. Read the codebase deeply first — your design\n"
        "should feel like a natural extension of what already exists.\n\n"
        f"Write your architecture document to: {architecture_path}\n\n"
        "The bar: this document is the single source of truth. Every interface you define\n"
        "will be copied verbatim into code.\n"
    )

from swe_af.execution.schemas import WorkspaceManifest
from swe_af.prompts._utils import workspace_context_block

def sprint_planner_task_prompt(
    *,
    goal: str,
    prd: "dict | PRD",
    architecture: "dict | Architecture",
    workspace_manifest: "WorkspaceManifest | None" = None,
    prd_path: str = "",
    architecture_path: str = "",
    repo_path: str = "",
) -> str:
    """Task prompt for the sprint planner agent.

    Builds the full sprint planner task prompt. Equivalent to
    sprint_planner_prompts()[1] with:
    - workspace_manifest keyword parameter added
    - Multi-repo mandate section injected when len(repos) > 1

    Multi-repo mandate ensures:
    - Each PlannedIssue sets target_repo to the repo it codes in
    - Cross-repo work is split into separate issues

    AC-18: 'target_repo' must appear in the returned string for multi-repo
    manifests (verified by the test). This is satisfied by the mandate section.
    """
    # Accept both Pydantic model and dict forms
    if hasattr(prd, "validated_description"):
        desc = prd.validated_description
        ac = prd.acceptance_criteria
    else:
        desc = prd.get("validated_description", goal)
        ac = prd.get("acceptance_criteria", [])

    if hasattr(architecture, "summary"):
        arch_summary = architecture.summary
    else:
        arch_summary = architecture.get("summary", "")

    ac_formatted = "\n".join(f"- {c}" for c in ac)

    ws_block = workspace_context_block(workspace_manifest)
    ws_section = f"\n{ws_block}\n" if ws_block else ""

    # Reference docs section
    ref_docs = ""
    if prd_path or architecture_path:
        ref_docs = "\n## Reference Documents\n"
        if prd_path:
            ref_docs += f"- Full PRD: {prd_path}\n"
        if architecture_path:
            ref_docs += f"- Architecture: {architecture_path}\n"

    repo_section = f"\n## Repository\n{repo_path}\n" if repo_path else ""

    # Multi-repo mandate — injected only when multiple repos present
    multi_repo_section = ""
    if workspace_manifest and len(workspace_manifest.repos) > 1:
        repo_names = [r.repo_name for r in workspace_manifest.repos]
        primary = workspace_manifest.primary_repo_name
        multi_repo_section = (
            "\n## Multi-Repo Mandate\n\n"
            "This build spans multiple repositories. For EVERY PlannedIssue:\n\n"
            f"1. **Set `target_repo`** to the repo name this issue codes in.\n"
            f"   Valid repo names: {repo_names}\n"
            f"   Primary repo (default): '{primary}'\n"
            "2. **One repo per issue**: do not create an issue that modifies files in\n"
            "   multiple repos simultaneously. If work spans repos, split into separate\n"
            "   issues with explicit `depends_on` edges.\n"
            "3. **Cross-repo dependencies**: an issue in repo B that consumes an interface\n"
            "   from repo A must list the repo-A issue in `depends_on`.\n"
        )

    return (
        f"## Goal\n{desc}\n\n"
        f"## Acceptance Criteria\n{ac_formatted}\n\n"
        f"## Architecture Summary\n{arch_summary}\n"
        f"{ws_section}"
        f"{ref_docs}"
        f"{repo_section}"
        f"{multi_repo_section}"
        "\n## Your Mission\n\n"
        "Break this work into issues executable by autonomous coder agents.\n\n"
        "Read the codebase, PRD, and architecture document thoroughly. The architecture\n"
        "document is your source of truth for all types, interfaces, and component\n"
        "boundaries.\n\n"
        "DO NOT write issue .md files. A separate parallel agent pool does that.\n\n"
        "For each issue provide: name, title, 2-3 sentence description (WHAT not HOW),\n"
        "depends_on, provides, files_to_create, files_to_modify, acceptance_criteria,\n"
        "testing_strategy, and guidance.\n\n"
        "Minimize the critical path. Maximize parallelism. Every PRD acceptance\n"
        "criterion must map to at least one issue.\n"
    )

from swe_af.execution.schemas import WorkspaceManifest
from swe_af.prompts._utils import workspace_context_block

def coder_task_prompt(
    issue: dict,
    worktree_path: str = "",
    feedback: str = "",
    iteration: int = 1,
    project_context: "dict | None" = None,
    memory_context: "dict | None" = None,
    workspace_manifest: "WorkspaceManifest | None" = None,   # NEW
    target_repo: str = "",                                    # NEW
) -> str:
    """Task prompt for the coder agent.

    New parameters:
        workspace_manifest: When provided for a multi-repo build, a CRITICAL
            directive block is prepended specifying which repo to code in.
        target_repo: Repo name (e.g. 'lib'). When set with a multi-repo
            manifest, the directive includes the absolute path of that repo.

    AC-19: When target_repo='lib' and manifest has lib at '/tmp/lib',
    the returned string contains '/tmp/lib'.
    """
    project_context = project_context or {}
    memory_context = memory_context or {}
    sections: list[str] = []

    # Multi-repo CRITICAL directive block — prepended before issue details
    if workspace_manifest and len(workspace_manifest.repos) > 1 and target_repo:
        target_path = ""
        for r in workspace_manifest.repos:
            if r.repo_name == target_repo:
                target_path = r.absolute_path
                break
        if not target_path:
            target_path = worktree_path  # fallback

        sections.append(
            f"## CRITICAL: Target Repository\n"
            f"You are working in the **{target_repo}** repository.\n"
            f"- Absolute path: `{target_path}`\n"
            "- ALL git operations (commit, status, add) must be run from this path.\n"
            "- Do NOT modify files in other repositories.\n"
        )
        ws_block = workspace_context_block(workspace_manifest)
        if ws_block:
            sections.append(ws_block)
        sections.append("")  # blank line separator

    # --- Existing body begins here (unchanged) ---
    sections.append("## Issue to Implement")
    sections.append(f"- **Name**: {issue.get('name', '(unknown)')}")
    sections.append(f"- **Title**: {issue.get('title', '(unknown)')}")

    ac = issue.get("acceptance_criteria", [])
    if ac:
        sections.append("- **Acceptance Criteria**:")
        sections.extend(f"  - [ ] {c}" for c in ac)

    deps = issue.get("depends_on", [])
    if deps:
        sections.append(f"- **Dependencies**: {deps}")

    provides = issue.get("provides", [])
    if provides:
        sections.append(f"- **Provides**: {provides}")

    files_create = issue.get("files_to_create", [])
    files_modify = issue.get("files_to_modify", [])
    if files_create:
        sections.append(f"- **Files to create**: {files_create}")
    if files_modify:
        sections.append(f"- **Files to modify**: {files_modify}")

    testing_strategy = issue.get("testing_strategy", "")
    if testing_strategy:
        sections.append(f"- **Testing Strategy**: {testing_strategy}")

    guidance = issue.get("guidance") or {}
    testing_guidance = guidance.get("testing_guidance", "")
    if testing_guidance:
        sections.append(f"- **Testing Guidance (from sprint planner)**: {testing_guidance}")

    if project_context:
        sections.append("\n## Project Context")
        prd_path = project_context.get("prd_path", "")
        arch_path = project_context.get("architecture_path", "")
        issues_dir = project_context.get("issues_dir", "")
        if prd_path or arch_path or issues_dir:
            sections.append("### Key Files")
            if prd_path:
                sections.append(f"- PRD: `{prd_path}`")
            if arch_path:
                sections.append(f"- Architecture: `{arch_path}`")
            if issues_dir:
                sections.append(f"- Issue files: `{issues_dir}/`")

    conventions = memory_context.get("codebase_conventions")
    if conventions:
        sections.append("\n## Codebase Conventions (from prior issues)")
        if isinstance(conventions, dict):
            for k, v in conventions.items():
                sections.append(f"- **{k}**: {v}")
        elif isinstance(conventions, list):
            sections.extend(f"- {c}" for c in conventions)

    failure_patterns = memory_context.get("failure_patterns")
    if failure_patterns:
        sections.append("\n## Known Failure Patterns (avoid these)")
        for fp in failure_patterns[:5]:
            sections.append(
                f"- **{fp.get('pattern', '?')}** ({fp.get('issue', '?')}): "
                f"{fp.get('description', '')}"
            )

    dep_interfaces = memory_context.get("dependency_interfaces")
    if dep_interfaces:
        sections.append("\n## Dependency Interfaces (completed upstream issues)")
        for iface in dep_interfaces:
            sections.append(f"- **{iface.get('issue', '?')}**: {iface.get('summary', '')}")
            exports = iface.get("exports", [])
            if exports:
                sections.extend(f"  - `{e}`" for e in exports[:5])

    bug_patterns = memory_context.get("bug_patterns")
    if bug_patterns:
        sections.append("\n## Common Bug Patterns in This Build")
        for bp in bug_patterns[:5]:
            sections.append(
                f"- {bp.get('type', '?')} (seen {bp.get('frequency', 0)}x "
                f"in {bp.get('modules', [])})"
            )

    failure_notes = issue.get("failure_notes", [])
    if failure_notes:
        sections.append("\n## Upstream Failure Notes")
        sections.extend(f"- {note}" for note in failure_notes)

    integration_branch = issue.get("integration_branch", "")
    if integration_branch:
        sections.append("\n## Git Context")
        sections.append(f"- Integration branch: `{integration_branch}`")
        sections.append(f"- Working in worktree: `{worktree_path}`")

    sections.append(f"\n## Working Directory\n`{worktree_path}`")
    sections.append(f"\n## Iteration: {iteration}")

    if feedback:
        sections.append("\n## Feedback from Previous Iteration")
        sections.append("Address ALL of the following issues from the review:\n")
        sections.append(feedback)
        sections.append(
            "\nFix the issues above, then re-commit. Focus on the specific "
            "problems identified — do not rewrite code that is already correct."
        )
    else:
        sections.append(
            "\n## Your Task\n"
            "1. Explore the codebase to understand patterns and context.\n"
            "2. Implement the solution per the acceptance criteria.\n"
            "3. Write or update tests per the Testing Strategy/guidance.\n"
            "4. Run tests and report results (tests_passed, test_summary).\n"
            "5. Commit your changes.\n"
            "6. Report codebase_learnings and agent_retro in your output."
        )

    return "\n".join(sections)

# ADD to existing verifier_task_prompt signature:
def verifier_task_prompt(
    prd: dict,
    artifacts_dir: str,
    completed_issues: list[dict],
    failed_issues: list[dict],
    skipped_issues: list[str],
    build_health: "dict | None" = None,
    workspace_manifest: "WorkspaceManifest | None" = None,  # NEW
) -> str:
    # ... existing body unchanged ...
    # ADD at end, before final return "\n".join(sections):
    ws_block = workspace_context_block(workspace_manifest)
    if ws_block:
        sections.append(f"\n{ws_block}")
    # return "\n".join(sections)

from swe_af.execution.schemas import WorkspaceManifest
from swe_af.prompts._utils import workspace_context_block

def workspace_setup_task_prompt(
    repo_path: str,
    integration_branch: str,
    issues: list[dict],
    worktrees_dir: str,
    build_id: str = "",
    workspace_manifest: "WorkspaceManifest | None" = None,  # NEW
) -> str:
    # ... existing body unchanged ...
    # ADD before final return:
    if workspace_manifest and len(workspace_manifest.repos) > 1:
        sections.append("\n## Per-Repo Paths")
        for r in workspace_manifest.repos:
            sections.append(f"- `{r.repo_name}` ({r.role}): `{r.absolute_path}`")
    return "\n".join(sections)

def github_pr_task_prompt(
    *,
    repo_path: str,
    integration_branch: str,
    base_branch: str,
    goal: str,
    build_summary: str = "",
    completed_issues: "list[dict] | None" = None,
    accumulated_debt: "list[dict] | None" = None,
    all_pr_results: "list[RepoPRResult] | None" = None,  # NEW
) -> str:
    # ... existing body unchanged ...
    # ADD cross-repo PR section when applicable:
    if all_pr_results and len(all_pr_results) > 1:
        sections.append("\n### Cross-Repo PRs")
        sections.append("Reference these sibling PRs in the PR body:")
        for r in all_pr_results:
            if r.pr_url:
                sections.append(f"- `{r.repo_name}`: {r.pr_url}")
    # ... rest unchanged ...

from swe_af.execution.schemas import RepoPRResult

from swe_af.prompts._utils import workspace_context_block

__all__ = [
    # ... existing exports ...
    "workspace_context_block",
    # new task prompt functions:
    "pm_task_prompt",
    "architect_task_prompt",
    "sprint_planner_task_prompt",
]

# REMOVE local definition:
# def _repo_name_from_url(url: str) -> str: ...

# ADD to imports:
from swe_af.execution.schemas import (
    BuildConfig, BuildResult,
    RepoPRResult, WorkspaceManifest,
    _derive_repo_name as _repo_name_from_url,
)

async def _clone_repos(
    cfg: BuildConfig,
    artifacts_dir: str,
) -> WorkspaceManifest:
    """Clone all repos from cfg.repos concurrently. Returns a WorkspaceManifest.

    Parameters:
        cfg: BuildConfig with .repos list populated. len(cfg.repos) >= 1.
        artifacts_dir: Absolute path used to derive workspace_root as its parent.

    Returns:
        WorkspaceManifest with one WorkspaceRepo per RepoSpec.
        All WorkspaceRepo.git_init_result fields are None at this stage
        (populated later by _init_all_repos in dag_executor.py).

    Raises:
        RuntimeError: If any git clone subprocess fails. Partially-cloned
            directories are removed (shutil.rmtree) before raising, so no
            orphaned workspace directories remain.

    Concurrency model:
        asyncio.gather([asyncio.to_thread(blocking_clone), ...]) for all N repos.
        Branch resolution also runs concurrently via asyncio.to_thread.
    """
    import shutil
    from swe_af.execution.schemas import WorkspaceRepo

    workspace_root = os.path.join(os.path.dirname(artifacts_dir), "workspace")
    os.makedirs(workspace_root, exist_ok=True)

    cloned_paths: list[str] = []

    async def _clone_single(spec: "RepoSpec") -> tuple[str, str]:
        """Clone or resolve one repo. Returns (repo_name, absolute_path)."""
        name = (
            spec.mount_point
            or (_repo_name_from_url(spec.repo_url) if spec.repo_url
                else os.path.basename(spec.repo_path.rstrip("/")))
        )
        dest = os.path.join(workspace_root, name)

        # If repo_path given, use it directly — no clone needed (resolves C-NEW-1)
        if spec.repo_path:
            return name, spec.repo_path

        git_dir = os.path.join(dest, ".git")
        if spec.repo_url and not os.path.exists(git_dir):
            os.makedirs(dest, exist_ok=True)
            cmd = ["git", "clone", spec.repo_url, dest]
            if spec.branch:
                cmd += ["--branch", spec.branch]

            def _run() -> subprocess.CompletedProcess:
                return subprocess.run(cmd, capture_output=True, text=True)

            proc = await asyncio.to_thread(_run)
            if proc.returncode != 0:
                raise RuntimeError(
                    f"git clone {spec.repo_url!r} failed "
                    f"(exit {proc.returncode}): {proc.stderr.strip()}"
                )
            cloned_paths.append(dest)

        return name, dest

    async def _resolve_branch(spec: "RepoSpec", path: str) -> str:
        """Resolve actual checked-out branch via git rev-parse (resolves C-NEW-1).

        Falls back to spec.branch or 'HEAD' on error.
        """
        if spec.repo_path and not spec.repo_url:
            # Existing repo — query actual branch
            pass
        def _run() -> str:
            r = subprocess.run(
                ["git", "-C", path, "rev-parse", "--abbrev-ref", "HEAD"],
                capture_output=True, text=True,
            )
            if r.returncode == 0 and r.stdout.strip():
                return r.stdout.strip()
            return spec.branch or "HEAD"
        return await asyncio.to_thread(_run)

    # Clone all repos concurrently
    clone_tasks = [_clone_single(spec) for spec in cfg.repos]
    clone_results = await asyncio.gather(*clone_tasks, return_exceptions=True)

    # Check for failures, cleanup partial clones
    errors = [
        (i, r) for i, r in enumerate(clone_results) if isinstance(r, Exception)
    ]
    if errors:
        for p in cloned_paths:
            shutil.rmtree(p, ignore_errors=True)
        msgs = "; ".join(str(r) for _, r in errors)
        raise RuntimeError(f"Multi-repo clone failed: {msgs}")

    # Resolve branches concurrently
    branch_tasks = [
        _resolve_branch(cfg.repos[i], clone_results[i][1])
        for i in range(len(cfg.repos))
    ]
    branches = await asyncio.gather(*branch_tasks, return_exceptions=True)

    # Build WorkspaceRepo list
    repos: list[WorkspaceRepo] = []
    primary_repo_name = ""

    for i, spec in enumerate(cfg.repos):
        name, path = clone_results[i]
        branch = branches[i] if isinstance(branches[i], str) else (spec.branch or "HEAD")
        ws_repo = WorkspaceRepo(
            repo_name=name,
            repo_url=spec.repo_url,
            role=spec.role,
            absolute_path=path,
            branch=branch,
            sparse_paths=spec.sparse_paths,
            create_pr=spec.create_pr,
            git_init_result=None,
        )
        repos.append(ws_repo)
        if spec.role == "primary":
            primary_repo_name = name

    return WorkspaceManifest(
        workspace_root=workspace_root,
        repos=repos,
        primary_repo_name=primary_repo_name,
    )

manifest: WorkspaceManifest | None = None

# Multi-repo path: clone all repos concurrently
if len(cfg.repos) > 1:
    app.note(
        f"Cloning {len(cfg.repos)} repos concurrently",
        tags=["build", "clone", "multi-repo"],
    )
    manifest = await _clone_repos(cfg, abs_artifacts_dir)
    # Use primary repo as the canonical repo_path
    repo_path = manifest.primary_repo.absolute_path
    app.note(
        f"Multi-repo workspace ready: {manifest.workspace_root}",
        tags=["build", "clone", "multi-repo", "complete"],
    )
# Single-repo path: existing clone logic unchanged (cfg.repos has 0 or 1 entry)

dag_result = _unwrap(await app.call(
    f"{NODE_ID}.execute",
    plan_result=plan_result,
    repo_path=repo_path,
    execute_fn_target=cfg.execute_fn_target,
    config=exec_config,
    git_config=git_config,
    build_id=build_id,
    workspace_manifest=manifest.model_dump() if manifest else None,  # NEW
), "execute")

pr_results: list[RepoPRResult] = []

if manifest and len(manifest.repos) > 1:
    # Multi-repo: one PR per repo where create_pr=True
    for ws_repo in manifest.repos:
        if not ws_repo.create_pr or not cfg.enable_github_pr:
            continue
        repo_git_init = ws_repo.git_init_result or {}
        repo_remote_url = repo_git_init.get("remote_url", "") or ws_repo.repo_url
        if not repo_remote_url:
            continue
        repo_integration_branch = repo_git_init.get("integration_branch", "")
        if not repo_integration_branch:
            continue
        repo_base_branch = (
            cfg.github_pr_base
            or repo_git_init.get("remote_default_branch", "")
            or "main"
        )
        try:
            pr_r = _unwrap(await app.call(
                f"{NODE_ID}.run_github_pr",
                repo_path=ws_repo.absolute_path,
                integration_branch=repo_integration_branch,
                base_branch=repo_base_branch,
                goal=goal,
                build_summary=build_summary,
                completed_issues=[
                    r for r in dag_result.get("completed_issues", [])
                    if not r.get("repo_name") or r.get("repo_name") == ws_repo.repo_name
                ],
                accumulated_debt=dag_result.get("accumulated_debt", []),
                artifacts_dir=plan_result.get("artifacts_dir", artifacts_dir),
                model=resolved["git_model"],
                permission_mode=cfg.permission_mode,
                ai_provider=cfg.ai_provider,
            ), "run_github_pr")
            pr_results.append(RepoPRResult(
                repo_name=ws_repo.repo_name,
                repo_url=ws_repo.repo_url,
                success=pr_r.get("success", False),
                pr_url=pr_r.get("pr_url", ""),
                pr_number=pr_r.get("pr_number", 0),
                error_message=pr_r.get("error_message", ""),
            ))
        except Exception as e:
            pr_results.append(RepoPRResult(
                repo_name=ws_repo.repo_name,
                repo_url=ws_repo.repo_url,
                success=False,
                error_message=str(e),
            ))
else:
    # Single-repo: existing PR logic, wrap result in RepoPRResult
    # ... (existing pr_url resolution code) ...
    # pr_url = <result of existing logic>
    if pr_url:
        pr_results.append(RepoPRResult(
            repo_name=_repo_name_from_url(cfg.repo_url) if cfg.repo_url else "repo",
            repo_url=cfg.repo_url,
            success=True,
            pr_url=pr_url,
            pr_number=pr_result.get("pr_number", 0) if 'pr_result' in dir() else 0,
        ))

return BuildResult(
    plan_result=plan_result,
    dag_state=dag_result,
    verification=verification,
    success=success,
    summary=...,
    pr_results=pr_results,  # replaces pr_url=pr_url
).model_dump()

@app.reasoner()
async def execute(
    plan_result: dict,
    repo_path: str,
    execute_fn_target: str = "",
    config: "dict | None" = None,
    git_config: "dict | None" = None,
    resume: bool = False,
    build_id: str = "",
    workspace_manifest: "dict | None" = None,  # NEW (resolves C-NEW-4)
) -> dict:
    """Execute a planned DAG with self-healing replanning."""
    from swe_af.execution.dag_executor import run_dag
    from swe_af.execution.schemas import ExecutionConfig

    effective_config = dict(config) if config else {}
    exec_config = ExecutionConfig(**effective_config) if effective_config else ExecutionConfig()

    if execute_fn_target:
        async def execute_fn(issue, dag_state):
            return await app.call(
                execute_fn_target,
                issue=issue,
                repo_path=dag_state.repo_path,
            )
    else:
        execute_fn = None

    # Build dag_state via existing _init_dag_state, then inject workspace_manifest
    # (run_dag calls _init_dag_state internally; we pass workspace_manifest through
    #  the dag_state after construction)
    state = await run_dag(
        plan_result=plan_result,
        repo_path=repo_path,
        execute_fn=execute_fn,
        config=exec_config,
        note_fn=app.note,
        call_fn=app.call,
        node_id=NODE_ID,
        git_config=git_config,
        resume=resume,
        build_id=build_id,
        workspace_manifest=workspace_manifest,  # NEW — passed to run_dag
    )
    return state.model_dump()

async def run_dag(
    plan_result: dict,
    repo_path: str,
    execute_fn: "Callable | None",
    config: ExecutionConfig,
    note_fn: "Callable | None" = None,
    call_fn: "Callable | None" = None,
    node_id: str = "swe-planner",
    git_config: "dict | None" = None,
    resume: bool = False,
    build_id: str = "",
    workspace_manifest: "dict | None" = None,  # NEW
) -> DAGState:
    # ... existing body ...
    # After _init_dag_state():
    dag_state.workspace_manifest = workspace_manifest

    # After git init (when workspace_manifest is set), run per-repo git init:
    if workspace_manifest and call_fn:
        await _init_all_repos(
            dag_state=dag_state,
            call_fn=call_fn,
            node_id=node_id,
            git_model=config.git_model,
            ai_provider=config.ai_provider,
            build_id=build_id,
        )
    # ... rest of run_dag unchanged ...

async def _init_all_repos(
    dag_state: DAGState,
    call_fn: Callable,
    node_id: str,
    git_model: str,
    ai_provider: str,
    permission_mode: str = "",
    build_id: str = "",
) -> None:
    """Run git_init concurrently for all repos in workspace_manifest.

    Resolves B-NEW-2 (ExecutionConfig clarified above) and B-NEW-4 (WorkspaceRepo
    mutability via ConfigDict(frozen=False)).

    The 'resolved' dict from build() is NOT threaded through here. Instead,
    the relevant model strings are passed directly as typed parameters:
      - git_model: from config.git_model (ExecutionConfig property)
      - ai_provider: from config.ai_provider (ExecutionConfig property)

    These are provided by run_dag which has access to the ExecutionConfig.

    Args:
        dag_state: Mutated in-place. dag_state.workspace_manifest must be a
            dict (WorkspaceManifest.model_dump()) set before calling.
        call_fn: AgentField call function for invoking run_git_init.
        node_id: e.g. 'swe-planner'.
        git_model: Resolved model string. Source: config.git_model.
        ai_provider: 'claude' or 'opencode'. Source: config.ai_provider.
        permission_mode: Forwarded to run_git_init. Source: dag_state has no
            permission_mode field — pass empty string for now; callers that
            need it can add it to run_dag's parameter list.
        build_id: Forwarded to run_git_init for branch namespace isolation.

    Postcondition:
        dag_state.workspace_manifest is updated with git_init_result populated
        on each WorkspaceRepo entry.
    """
    if dag_state.workspace_manifest is None:
        return  # single-repo path: git_init already ran in build()

    from swe_af.execution.schemas import WorkspaceManifest, WorkspaceRepo

    manifest = WorkspaceManifest(**dag_state.workspace_manifest)

    async def _init_one(ws_repo: WorkspaceRepo) -> tuple[str, dict]:
        result = await call_fn(
            f"{node_id}.run_git_init",
            repo_path=ws_repo.absolute_path,
            goal="",  # goal not needed for dependency repos
            artifacts_dir=dag_state.artifacts_dir,
            model=git_model,
            permission_mode=permission_mode,
            ai_provider=ai_provider,
            build_id=build_id,
        )
        return ws_repo.repo_name, result

    tasks = [_init_one(r) for r in manifest.repos]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    # Write results back (WorkspaceRepo is mutable: model_config = ConfigDict(frozen=False))
    repo_map = {r.repo_name: r for r in manifest.repos}
    for item in results:
        if isinstance(item, Exception):
            if note_fn:  # note_fn may not be available in this helper
                pass
            continue  # non-fatal: single-repo git_init failure is already non-fatal
        name, git_init_dict = item
        if name in repo_map:
            repo_map[name].git_init_result = git_init_dict

    # Replace dag_state manifest dict with updated version
    dag_state.workspace_manifest = manifest.model_dump()

# When recording a successful coding loop result:
return IssueResult(
    issue_name=issue["name"],
    outcome=IssueOutcome.COMPLETED,
    result_summary=coder_result.summary,
    files_changed=coder_result.files_changed,
    branch_name=issue.get("branch_name", ""),
    repo_name=coder_result.repo_name,   # propagated from CoderResult
    # ... other fields ...
)

for i, result in enumerate(results):
    if isinstance(result, IssueResult):
        # Backfill repo_name from issue's target_repo if CoderResult didn't set it
        if not result.repo_name:
            result.repo_name = active_issues[i].get("target_repo", "")
        # ... existing classification logic ...

async def _merge_level_branches(
    dag_state: DAGState,
    level_result: LevelResult,
    call_fn: Callable,
    node_id: str,
    config: ExecutionConfig,
    issue_by_name: dict,
    file_conflicts: list[dict],
    note_fn: "Callable | None" = None,
) -> "dict | None":
    """Merge completed branches, dispatching per-repo when manifest is present."""

    # Single-repo path: unchanged
    if dag_state.workspace_manifest is None:
        # ... existing _merge_level_branches body verbatim ...
        pass

    # Multi-repo path: group by repo_name, one merger call per repo
    from swe_af.execution.schemas import WorkspaceManifest
    manifest = WorkspaceManifest(**dag_state.workspace_manifest)

    # Group IssueResults by repo_name (fall back to primary if empty)
    by_repo: dict[str, list] = {}
    for r in level_result.completed:
        if r.branch_name:
            repo = r.repo_name or manifest.primary_repo_name
            by_repo.setdefault(repo, []).append(r)

    if not by_repo:
        return None

    async def _call_merger_for_repo(
        repo_name: str,
        issue_results: list,
    ) -> "dict":
        """Invoke run_merger for a single repo. config provides all model fields."""
        ws_repo = next(
            (r for r in manifest.repos if r.repo_name == repo_name), None
        )
        if ws_repo is None or ws_repo.git_init_result is None:
            return {"success": False, "merged_branches": [], "failed_branches": []}

        git_init = ws_repo.git_init_result
        integration_branch = git_init.get("integration_branch", "")
        if not integration_branch:
            return {"success": False, "merged_branches": [], "failed_branches": []}

        branches_to_merge = [
            {
                "branch_name": r.branch_name,
                "issue_name": r.issue_name,
                "result_summary": r.result_summary,
                "files_changed": r.files_changed,
                "issue_description": issue_by_name.get(r.issue_name, {}).get("description", ""),
            }
            for r in issue_results
        ]

        # Uses config.merger_model and config.ai_provider (ExecutionConfig properties)
        result = await call_fn(
            f"{node_id}.run_merger",
            repo_path=ws_repo.absolute_path,
            integration_branch=integration_branch,
            branches_to_merge=branches_to_merge,
            file_conflicts=file_conflicts,
            prd_summary=dag_state.prd_summary,
            architecture_summary=dag_state.architecture_summary,
            artifacts_dir=dag_state.artifacts_dir,
            level=level_result.level_index,
            model=config.merger_model,
            ai_provider=config.ai_provider,
        )
        return result

    # Dispatch all repo merges concurrently
    tasks = [
        _call_merger_for_repo(repo_name, issues)
        for repo_name, issues in by_repo.items()
    ]
    repo_names = list(by_repo.keys())
    results = await asyncio.gather(*tasks, return_exceptions=True)

    last_good: dict | None = None
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            continue
        dag_state.merge_results.append({**result, "repo_name": repo_names[i]})
        for b in result.get("merged_branches", []):
            if b not in dag_state.merged_branches:
                dag_state.merged_branches.append(b)
        for b in result.get("failed_branches", []):
            if b not in dag_state.unmerged_branches:
                dag_state.unmerged_branches.append(b)
        if result.get("success"):
            last_good = result

    return last_good

BuildConfig(repo_url="https://github.com/org/repo.git")
# → _normalize_repos synthesizes repos=[RepoSpec(repo_url=..., role='primary')]
# → cfg.repo_url unchanged, cfg.primary_repo returns the single RepoSpec
# → build() sees len(cfg.repos)==1, takes existing single-repo path
# → BuildResult.pr_url property returns the same string as before

result = BuildResult(..., pr_results=[RepoPRResult(..., pr_url="https://...")])
result.pr_url           # "https://..." (property)
result.model_dump()["pr_url"]  # "https://..." (injected by model_dump override)